package com.atguigu.flink.keyStateAndOpeState;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class OperatorBroadcastState {
    public static void main(String[] args) throws Exception {
        // TODO 1.环境准备
        // 1.1指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2设置并行度
        env.setParallelism(1);
        // TODO 2.从指定的网络端口读取水位流信息
        SingleOutputStreamOperator<WaterSensor> ds1 = env
                .socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction());
        // TODO 3.从网络端口读取阈值信息
        SingleOutputStreamOperator<Integer> ds2 = env
                .socketTextStream("hadoop102", 8889)
                .map(Integer::parseInt);
        // TODO 4.将阈值信息流进行广播 -- 广播流
        MapStateDescriptor<String, Integer> mapStateDescriptor = new MapStateDescriptor<>("broadcast-state", String.class, Integer.class);
        BroadcastStream<Integer> broadcastDS = ds2.broadcast(mapStateDescriptor);
        // TODO 5.将水位流和广播流进行关联
        BroadcastConnectedStream<WaterSensor, Integer> connectDS = ds1.connect(broadcastDS);
        // TODO 6.对关联的数据进行处理
        SingleOutputStreamOperator<String> processDS = connectDS.process(
                new BroadcastProcessFunction<WaterSensor, Integer, String>() {

                    //非广播流
                    @Override
                    public void processElement(WaterSensor ws, BroadcastProcessFunction<WaterSensor, Integer, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        Integer vc = ws.getVc();
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        Integer threshold = broadcastState.get("threshold");
                        if (vc > threshold){
                            out.collect("当前水位值" + vc + "是超过了阈值" + threshold + "，赶快撤离");
                        }
                    }

                    // 广播流
                    @Override
                    public void processBroadcastElement(Integer value, BroadcastProcessFunction<WaterSensor, Integer, String>.Context ctx, Collector<String> out) throws Exception {
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        broadcastState.put("threshold" , value);
                    }
                }
        );
        // TODO 7.打印输出
        processDS.print();
        // TODO 8.提交作业
        env.execute();

    }
}
